package com.allwe.client.reBalance;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 生产者多线程安全 - 测试demo
 *
 * @Author: AllWe
 * @Date: 2024/09/27/9:30
 */
@Data
@Slf4j
public class ConcurrentProducerWorker {

    /**
     * 消息数量
     */
    private static final int RECORD_COUNT = 1000;

    /**
     * 固定线程池 - 线程数等于CPU核数
     */
    private static final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    /**
     * 发令枪
     */
    private static final CountDownLatch countDownLatch = new CountDownLatch(RECORD_COUNT);

    /**
     * 生产者 - 这里让所有的线程都共享同一个生产者
     */
    private static KafkaProducer<String, String> kafkaProducer;

    /**
     * 类初始化的时候 - 创建生产者实例
     */
    static {
        // 设置属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "39.107.66.31:9092");
        properties.put("key.serializer", StringSerializer.class);
        properties.put("value.serializer", StringSerializer.class);
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);

        kafkaProducer = new KafkaProducer<>(properties);
    }

    /**
     * 启动器
     */
    public static void main(String[] args) {
        try {
            // 循环创建消息
            for (int count = 0; count < RECORD_COUNT; count++) {
                ProducerRecord<String, String> record = new ProducerRecord<>("reBalance--1", "allwe", "allwe_" + count % 4);
                executorService.submit(new ConcurrentProducer(record, kafkaProducer, countDownLatch));
                Thread.sleep(1000);
            }
            countDownLatch.await();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者连接
            kafkaProducer.close();
            // 释放线程池资源
            executorService.shutdown();
        }
    }
}
